package com.yanggu.flink.datastream_api.partition

import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._


/**
 * rescale是会将数据轮询发送到下游并行任务的一部分中
 * rebalance是上游的所有子任务都与下游的子任务建立联系, 然后轮询发送数据, 是一种笛卡尔积的关系(M * N)
 * 如果下游算子的并行度是上游的算子的整数倍, 那么只是上游算子只是与下游的部分算子建立联系, 然后轮询发送数据
 * 具体为上游并行度为2(source0、source1), 下游并行度为4(rescale1、rescale2、rescale3、rescale4)
 * source0中的数据为2、4、6、8
 * source1中的数据为1、3、5、7
 * 上游source0子任务和下游rescale1子任务、下游rescale2子任务建立联系, 同时轮询发送数据(2、4、6、8)
 * 上游source1子任务和下游rescale3子任务、下游rescale4子任务建立联系, 同时轮询发送数据(1、3、5、7)
 *
 */
//source:0> 2、4、6、8
//source:1> 1、3、5、7
//rescale:1> 2    rescale:2> 4    rescale:1> 6    rescale:2> 8
//rescale:3> 1    rescale:4> 3    rescale:3> 5    rescale:4> 7
object RescaleDemo {

  def main(args: Array[String]): Unit = {
    // 创建执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 这里使用了并行数据源的富函数版本
    // 这样可以调用 getRuntimeContext 方法来获取运行时上下文的一些信息
    env.addSource(new RichParallelSourceFunction[Int] {
      override def run(ctx: SourceFunction.SourceContext[Int]) = {
        for (i <- 0 until 8) {
          // 将奇数发送到索引为 1 的并行子任务
          // 将偶数发送到索引为 0 的并行子任务
          if ((i + 1) % 2 == getRuntimeContext.getIndexOfThisSubtask) {
            ctx.collect(i + 1)
          }
        }
      }

      override def cancel() = {}

    }).setParallelism(2)
      .rescale
      .print("rescale").setParallelism(4)

    env.execute()
  }

}
